package com.jiajia.study.chapter01.nio;

import com.jiajia.study.utils.ByteBufferUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zjiajia
 * @date 2024/7/5 14:11
 */
@Slf4j
public class TestServerV5 {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");


        Selector boss = Selector.open();

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);

        SelectionKey serverSocketSelectionKey = serverSocketChannel.register(boss, 0, null);
        serverSocketSelectionKey.interestOps(SelectionKey.OP_ACCEPT);

        serverSocketChannel.bind(new InetSocketAddress(8888));
        int workerLength = 2;
        // 初始化 worker
        Worker[] workers = new Worker[workerLength];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }
        AtomicInteger number = new AtomicInteger(0);

        while(true){
            boss.select();
            Set<SelectionKey> selectionKeys = boss.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while(iterator.hasNext()){
                SelectionKey selectionKey = iterator.next();
                if(selectionKey.isAcceptable()){
                    iterator.remove();
                    serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel socketChannel = serverSocketChannel.accept();

                    socketChannel.configureBlocking(false);
                    log.debug("connected...{}", socketChannel.getRemoteAddress());
                    // 2. 关联 selector
                    log.debug("before register...{}", socketChannel.getRemoteAddress());
                    workers[number.getAndIncrement() % workers.length].register(socketChannel);
                    log.debug("after reggister...{}",socketChannel.getRemoteAddress());
                }

            }
        }


    }

    static class Worker implements Runnable{
        private String name;
        private Thread thread;
        private Selector selector;
        private volatile boolean start = false;

    public Worker(String name){
            this.name = name;
        }

        public void register(SocketChannel socketChannel) throws IOException {
            if(!start){
                selector = Selector.open();
                // 开启一个线程
                thread = new Thread(this,name);
                thread.start();
                start = true;
            }
            /**
             * 唤醒 selector.select 的阻塞
             *
             * 避免出现：
             *   一直阻塞在 selector,然后新的监听一直卡在这
             */
            selector.wakeup();
            // channel 的 读写事件监听给这个线程
            ByteBuffer byteBuffer = ByteBuffer.allocate(16);
            socketChannel.register(selector, SelectionKey.OP_READ,byteBuffer);
        }


        @Override
        public void run() {
            while(true){
                try {
                    /**
                     * 监听事件的发生
                     */
                    selector.select();

                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    if(iterator.hasNext()){
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                        try{
                            if(selectionKey.isReadable()){
                                ByteBuffer byteBuffer = (ByteBuffer)selectionKey.attachment();
                                log.debug("当前处理线程：{},read...{}",Thread.currentThread().getName(), socketChannel.getRemoteAddress());
                                int read = socketChannel.read(byteBuffer);
                                ByteBufferUtil.debugAll(byteBuffer);
                                if(read == -1 ){
                                    selectionKey.cancel();
                                }else {
                                    split(byteBuffer);
                                    // 没有空间可以写入数据了，说明需要扩容了
                                    if(byteBuffer.position() == byteBuffer.limit()){
                                        ByteBuffer newByteBuffer = ByteBuffer.allocate(byteBuffer.capacity() * 2);
                                        byteBuffer.flip();
                                        newByteBuffer.put(byteBuffer);
                                        // 放入扩容后的 ByteBuffer
                                        ByteBufferUtil.debugAll(newByteBuffer);
                                        selectionKey.attach(newByteBuffer);
                                    }

                                }
                            }else if(selectionKey.isWritable()){
                                // todo 写数据
                            }
                        }catch (Exception e){
                            e.printStackTrace();
                            /**
                             * 异常关闭，取消监听
                             */
                            selectionKey.cancel();
                        }


                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


        /**
         * 这种方式的效率比较低，需要一个一个的字节比较
         * @param byteBuffer    需要粘包粘包处理的 ByteBuffer
         */
        public  static void split(ByteBuffer byteBuffer){
            // 切换成读模式
            byteBuffer.flip();
            for (int i = 0; i < byteBuffer.limit(); i++) {
                if('\n' == (byteBuffer.get(i))){
                    int length = i+1 - byteBuffer.position();
                    ByteBuffer target = ByteBuffer.allocate(length);
                    for (int j = 0; j < length; j++) {
                        target.put(byteBuffer.get());
                    }
                    ByteBufferUtil.debugAll(target);
                }
            }
            byteBuffer.compact();
        }

    }

}

